package com.hr.reactive.flux;


import com.hr.reactive.common.Utils;
import reactor.core.publisher.Flux;

import java.util.ArrayList;

/**
 * FluxSink 线程安全
 *
 * @author Administrator
 */
public class Lec09FluxSinkThreadSafety {
    public static void main(String[] args) {
        demo2();
    }

    private static void demo1() {
        var list = new ArrayList<>();
        Runnable runnable = () -> {
            for (int i = 0; i < 1000; i++) {
                list.add(i);
            }
        };
        for (int i = 0; i < 10; i++) {
            new Thread(runnable).start();
        }
        Utils.sleepSeconds(5000);
        System.out.println(list.size());
    }

    //flux sink线程安全
    //ArrayList not thread safe
    private static void demo2() {
        var list = new ArrayList<String>();
        var g = new Lec08FluxCreate.NameGenerator();
        Flux<String> stringFlux = Flux.create(g);
        //线程安全
        stringFlux.subscribe(list::add);
        Runnable runnable = () -> {
            for (int i = 0; i < 1000; i++) {
                g.generator();
            }
        };
        for (int i = 0; i < 10; i++) {
            new Thread(runnable).start();
        }
        System.out.println(list.size());
        Utils.sleepSeconds(5000);
        System.out.println(list.size());
    }
}
